﻿
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Diagnostics;

namespace WSU.ChekaParallel.TestPLINQ
{
    static class TestBasicPLINQ
    {
        private static void PrintInSequence<TData>(this IEnumerable<TData> sequence)
        {
            foreach (TData data in sequence)
            {
                Console.WriteLine("Thread<{0}>: {1}", Thread.CurrentThread.ManagedThreadId, data);
            }
        }

        /// <summary>
        /// chekanote: demonstrate the usage of "ForAll"
        /// by using "ForAll", it bypass the procedure of collating subresult from different threads together
        /// so delegate passed into "ForAll" will be executed on different threads
        /// an optimization when there are a lot of elements to enumerate and you don't care about their order
        /// </summary>
        private static void PrintInParallel<TData>(this IEnumerable<TData> sequence)
        {
            ParallelQuery<TData> parallelSequence = sequence as ParallelQuery<TData>;
            if (parallelSequence == null)
                throw new InvalidOperationException("input sequence is not a parallel query");
            parallelSequence.ForAll(element =>
            {
                Console.WriteLine("Thread<{0}>: {1}", Thread.CurrentThread.ManagedThreadId, element);
            });
        }

        private static void SampleTest1()
        {
            // chekanote: force using range partitioning
            // you can also use "inputs = new int[100]", but that will cost extra memory
            IEnumerable<int> inputs = Enumerable.Range(1, 100).ToArray();

            // execute once and collate results together, and then print them in a single thread
            IEnumerable<int> filtered = from num in inputs.AsParallel()
                                        where num % 9 == 0
                                        select num;
            filtered.PrintInSequence();
            Console.WriteLine("************ first run completes, press enter to start next run,...");
            Console.ReadLine();

            // chekawarn: parallel query is also deferred
            // but here we just execute the same query, because the parallel query is executed only once
            // when "AsParallel" is executed, re-assign the original input will not automatically re-create the parallel query
            inputs = Enumerable.Range(2, 59).ToArray();
            filtered.PrintInParallel();
        }

        private static void TestDeferredFeature()
        {
            IEnumerable<int> inputs = null;
            Func<IEnumerable<int>> makeParallelQuery = () => from num in inputs.AsParallel()
                                                             where num % 9 == 0
                                                             select num * 2;
            inputs = Enumerable.Range(2, 40).ToArray();
            makeParallelQuery().PrintInParallel();
        }

        private static int SimulatedHeavyCalculation(int num, int sleepTime = 1000)
        {
            Thread.Sleep(sleepTime);
            return num * 2;
        }

        /// <summary>
        /// if consuming the parallel query by "foreach"
        /// then cancel the parallel query is very simple by just "break"
        /// </summary>
        private static void TestCancelByBreak()
        {
            // chekanote: what ParallelEnumerable.Range returns is already "ParallelQuery", no need to append "AsParallel" operation
            IEnumerable<int> results = from num in ParallelEnumerable.Range(1, 100)
                                       where num % 9 == 0
                                       select SimulatedHeavyCalculation(num);

            int total = 0;
            foreach (int result in results)
            {
                ++total;
                Console.WriteLine(result);

                if (total >= 3)
                    break;
            }
        }

        /// <summary>
        /// this cancellation method is used when the result sequence will not be consumed in "foreach" loop
        /// and there is no chance to use "break"
        /// </summary>
        private static void TestCancelByToken(int sleepTime)
        {
            CancellationTokenSource cancelSource = new CancellationTokenSource();

            IEnumerable<int> results = from num in ParallelEnumerable.Range(1, 100).WithCancellation(cancelSource.Token)
                                       where num % 11 == 0
                                       select SimulatedHeavyCalculation(num, sleepTime);
            Debug.Assert(results is ParallelQuery<int>);

            // ------------- start a timer to cancel the query after a specific period
            System.Timers.Timer timer = new System.Timers.Timer { Interval = 2000, AutoReset = false };
            timer.Elapsed += (sender, evtargs) =>
                                 {
                                     Console.WriteLine("timer out, cancel the query.");
                                     cancelSource.Cancel();
                                 };
            timer.Start();

            try
            {
                int[] resultArray = results.ToArray();
                Console.WriteLine("Results successfully got.");
            }
            catch (OperationCanceledException)// NOT AggregateException
            {
                Console.WriteLine("!!! Canceled Exception Caught !!!");
            }

        }

        private static void TestParallelAggregate()
        {
            int start = 1;
            int size = 100;
            int expectedTotal = start * size + size * (size - 1) / 2;

            // --------------- do the parallel aggregation
            int parallelResult = ParallelEnumerable.Range(start, size).Aggregate(
                () => 0,
                (localTotal, num) => localTotal + num,
                (mainTotal, localTotal) => mainTotal + localTotal,
                finalResult => finalResult);

            Debug.Assert(parallelResult == expectedTotal);
            Console.WriteLine("parallel aggregation done");
        }

        private static void TestParallelAggregate2()
        {
            string text = "abcdefghijklmnopqrstuvwxyzABCD";

            int[] statistics = text.AsParallel().Aggregate(
                () => new int[26],
                (localFreq, c) =>
                {
                    int index = char.ToUpper(c) - 'A';
                    if (index >= 0 && index <= 26)
                        localFreq[index]++;
                    return localFreq;
                },
                // pay attention to "ToArray", the merged result must be concreted
                // NO deferred execution feature should be used here, because the local accumulator may be disposed
                // when this merging operation is finished
                (mainFreq, localFreq) => mainFreq.Zip(localFreq, (f1, f2) => f1 + f2).ToArray(),
                finalresult => finalresult
                );

            for (int index = 0; index < statistics.Length; ++index)
            {
                char letter = (char)((int)'A' + index);
                Console.WriteLine("{0}:{1}", letter, statistics[index]);
            }
        }

        public static void TestMain()
        {
            // SampleTest1();
            // TestDeferredFeature();
            // TestCancelByBreak();
            // TestCancelByToken(10);// completes before being canceled
            // TestCancelByToken(1000);// canceled before completes
            // TestParallelAggregate();
            TestParallelAggregate2();
        }
    }
}